package cn.kuporo.seckill.goods.application.cacahe.impl;

import cn.kuporo.seckill.common.cache.distribute.DistributedCacheService;
import cn.kuporo.seckill.common.cache.local.caffeine.LocalCacheFactory;
import cn.kuporo.seckill.common.cache.model.SeckillBusinessCache;
import cn.kuporo.seckill.common.constants.RedisConstants;
import cn.kuporo.seckill.common.lock.DistributedLock;
import cn.kuporo.seckill.common.lock.factory.DistributedLockFactory;
import cn.kuporo.seckill.common.model.dto.SeckillStockDTO;
import cn.kuporo.seckill.common.model.enums.PlaceOrderType;
import cn.kuporo.seckill.common.utils.SystemClock;
import cn.kuporo.seckill.dubbo.interfaces.stock.SeckillStockDubboService;
import cn.kuporo.seckill.goods.application.cacahe.SeckillGoodsCacheService;
import cn.kuporo.seckill.goods.domain.model.SeckillGoods;
import cn.kuporo.seckill.goods.domain.service.SeckillGoodsDomainService;
import com.alibaba.fastjson2.TypeReference;
import com.github.benmanes.caffeine.cache.Cache;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.DubboReference;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author: kuporo
 * @since: 2025/7/9
 * @description:
 */
@Service
@Slf4j
@RequiredArgsConstructor
public class SeckillGoodsCacheServiceImpl implements SeckillGoodsCacheService {
    // private final LocalCacheService<Long, SeckillBusinessCache<SeckillGoods>> localCacheService;
    private static final Cache<Long, SeckillBusinessCache<SeckillGoods>> localSeckillGoodsCacheService = LocalCacheFactory.getLocalCache();
    private static final Cache<Long, SeckillBusinessCache<Integer>> localAvailableStockCacheService = LocalCacheFactory.getLocalCache();
    
    
    private final DistributedCacheService distributedCacheService;
    private final SeckillGoodsDomainService seckillGoodsDomainService;
    private final DistributedLockFactory distributedLockFactory;
    @DubboReference
    private SeckillStockDubboService seckillStockDubboService;
    @Value("${place.order.type:lua}")
    private String placeOrderType;
    /**
     * 本地锁
     */
    private final Lock localCacheUpdatelock = new ReentrantLock();
    @Override
    public String buildCacheKey(Object key) {
        return RedisConstants.SECKILL_GOODS_CACHE_KEY + key;
    }

    @Override
    public String buildLockKey(Object key) {
        return RedisConstants.SECKILL_GOODS_UPDATE_CACHE_LOCK_KEY + key;
    }

    @Override
    public SeckillBusinessCache<SeckillGoods> getSeckillGoods(Long goodsId, Long version) {
        // 1.获取本地缓存
        SeckillBusinessCache<SeckillGoods> seckillGoodsLocalCache = localSeckillGoodsCacheService.getIfPresent(goodsId);
        if (seckillGoodsLocalCache != null){
            // 1.1.如果版本号为空，直接返回
            if (version == null){
                log.info("SeckillGoodsCache|命中本地缓存:[版本号为null]|商品id:{}", goodsId);
                return seckillGoodsLocalCache;
            }
            // 1.2.传递过来的版本小于或等于缓存中的版本号，直接返回
            if (version.compareTo(seckillGoodsLocalCache.getVersion()) <= 0){
                log.info("SeckillGoodsCache|命中本地缓存:[版本号<缓存版本号]|商品id:{}", goodsId);
                return seckillGoodsLocalCache;
            }
            if (version.compareTo(seckillGoodsLocalCache.getVersion()) > 0){
                log.info("SeckillGoodsCache|命中本地缓存:[版本号>缓存版本号]|商品id:{}", goodsId);
                return getDistributedCache(goodsId);
            }
        }
        log.info("SeckillGoodsCache|未命中本地缓存]|商品id:{}", goodsId);
        // 2.获取分布式缓存
        return getDistributedCache(goodsId);
    }

    private SeckillBusinessCache<SeckillGoods> getDistributedCache(Long goodsId) {
        log.info("SeckillGoodsCachev|读取分布式缓存|商品id:{}", goodsId);
        // 2.1.查询分布式缓存
        SeckillBusinessCache<SeckillGoods> seckillgoodsDistributeCache = distributedCacheService.getGenericObject(buildCacheKey(goodsId), new TypeReference<SeckillBusinessCache<SeckillGoods>>() {
        });
        if (seckillgoodsDistributeCache == null){
            // 2.2.分布式缓存为空，尝试使用1个线程更新分布式缓存中的数据
            log.info("SeckillGoodsCache|分布式缓存:[未命中]|商品id:{}", goodsId);
            seckillgoodsDistributeCache = tryUpdateSeckillGoodsCacheByLock(goodsId, true);
        }
        if (seckillgoodsDistributeCache != null && !seckillgoodsDistributeCache.isRetryLater()){
            // 2.3.分布式缓存不为空，且不需要进行重试
            if (localCacheUpdatelock.tryLock()){
                try {
                    localSeckillGoodsCacheService.put(goodsId, seckillgoodsDistributeCache);
                    log.info("SeckillGoodsCache|本地缓存:[更新完成]:商品id:{}", goodsId);
                }finally {
                    localCacheUpdatelock.unlock();
                }
            }
        }
        return seckillgoodsDistributeCache;
    }

    @Override
    public SeckillBusinessCache<SeckillGoods> tryUpdateSeckillGoodsCacheByLock(Long goodsId, boolean doubleCheck) {
        String lockKey = buildLockKey(goodsId);
        long start = System.currentTimeMillis();
        //获取分布式锁，保证只有一个线程在更新分布式缓存
        DistributedLock lock = distributedLockFactory.getDistributedLock(lockKey);
        try {
            boolean isLockSuccess = lock.tryLock(2, 5, TimeUnit.MILLISECONDS);
            if (!isLockSuccess){
                // 没有获取到分布式锁，返回重试数据模型
                return new SeckillBusinessCache<SeckillGoods>().retryLater();
            }
            log.info("SeckillGoodsCache|分布式缓存:[更新中]|商品id{}", goodsId);
            SeckillBusinessCache<SeckillGoods> seckillGoodsDistributeCache;
            if (doubleCheck){
                //获取锁成功后，再次从缓存中获取数据，防止高并发下多个线程争抢锁的过程中，后续的线程再等待1秒的过程中，前面的线程释放了锁，后续的线程获取锁成功后再次更新分布式缓存数据
                seckillGoodsDistributeCache = distributedCacheService.getGenericObject(buildCacheKey(goodsId), new TypeReference<SeckillBusinessCache<SeckillGoods>>() {
                });
                if (seckillGoodsDistributeCache != null){
                    return seckillGoodsDistributeCache;
                }
            }
            SeckillGoods seckillGoods = seckillGoodsDomainService.getSeckillGoodsId(goodsId);
            if (seckillGoods == null){
                // 查询的数据为空，返回空数据模型
                seckillGoodsDistributeCache = new SeckillBusinessCache<SeckillGoods>().notExist();
            }else {
                // 数据存在，返回数据+版本号模型
                seckillGoodsDistributeCache = new SeckillBusinessCache<SeckillGoods>().with(this.getSeckillGoods(seckillGoods)).withVersion(SystemClock.millisClock().now());
            }
            distributedCacheService.put(buildCacheKey(goodsId), seckillGoodsDistributeCache, RedisConstants.FIVE_MINUTES);
            log.info("SeckillGoodsCache|分布式缓存:[更新成功]-商品id{}", goodsId);
            return seckillGoodsDistributeCache;
        }catch (InterruptedException e){
            // 分布式缓存更新失败，返回重试模型
            log.info("SeckillGoodsCache|分布式缓存:[更新失败]-商品id{}", goodsId);
            return new SeckillBusinessCache<SeckillGoods>().retryLater();
        }finally {
            log.info("{}释放锁，持有时间:{}",Thread.currentThread(), System.currentTimeMillis() - start);
            lock.unlock();
        }
    }

    @Override
    public SeckillBusinessCache<Integer> getAvailableStock(Long goodsId, Long version) {
        SeckillBusinessCache<Integer> availableStockCache = localAvailableStockCacheService.getIfPresent(goodsId);
        if (availableStockCache != null){
            //版本号为空，则直接返回本地缓存中的数据
            if (availableStockCache.getVersion() == null){
                log.info("availableStockCache|命中本地缓存|{}", goodsId);
                return availableStockCache;
            }
            //传递的版本号小于等于缓存中的版本号，则说明缓存中的数据比客户端的数据新，直接返回本地缓存中的数据
            if (version.compareTo(availableStockCache.getVersion()) <= 0){
                log.info("availableStockCache|命中本地缓存|{}", goodsId);
                return availableStockCache;
            }
            //传递的版本号大于缓存中的版本号，说明缓存中的数据比较落后，从分布式缓存获取数据并更新到本地缓存
            if (version.compareTo(availableStockCache.getVersion()) > 0){
                return getDistributedAvailableCache(goodsId);
            }
        }
        return getDistributedAvailableCache(goodsId);
    }

    /**
     * 从分布式缓存获取数据
     */
    private SeckillBusinessCache<Integer> getDistributedAvailableCache(Long goodsId) {
        log.info("SeckillGoodsCache|读取分布式缓存|{}", goodsId);
        //从分布式缓存中获取数据
        String goodsKey = RedisConstants.getKey(RedisConstants.GOODS_ITEM_STOCK_KEY_PREFIX, String.valueOf(goodsId));
        //获取商品库存
        Integer availableStock = distributedCacheService.getObject(goodsKey, Integer.class);
        if (availableStock == null){
            availableStock = seckillGoodsDomainService.getAvailableStockById(goodsId);
            distributedCacheService.put(goodsKey, availableStock);
        }
        SeckillBusinessCache<Integer> availableStockCache = new SeckillBusinessCache<Integer>().with(availableStock).withVersion(SystemClock.millisClock().now());
        localAvailableStockCacheService.put(goodsId, availableStockCache);
        return availableStockCache;
    }

    /**
     * 兼容分桶库存
     */
    private SeckillGoods getSeckillGoods(SeckillGoods seckillGoods) {
        // 不是分桶库存模式
        if (!PlaceOrderType.BUCKET.getType().equals(placeOrderType)){
            return seckillGoods;
        }
        // 是分桶库存模式，获取分桶库存
        SeckillBusinessCache<SeckillStockDTO> seckillStockCache = seckillStockDubboService.getSeckillStock(seckillGoods.getId(), 1L);
        if (seckillStockCache == null || !seckillStockCache.isExist() || seckillStockCache.isRetryLater() || seckillStockCache.getData() == null){
            return seckillGoods;
        }
        seckillGoods.setInitialStock(seckillStockCache.getData().getTotalStock());
        seckillGoods.setAvailableStock(seckillStockCache.getData().getAvailableStock());
        return seckillGoods;
    }
}
